package day03;

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.Collector;

import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * @Description: 连续1s数值上升
 * @Author: ZYX
 * @Date: 2022/2/10 15:57
 * @Version: 1.0
 */
public class Demo06 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        env.addSource(new SourceFunction<Integer>() {
            private boolean running = true;
            private Random random = new Random();

            @Override
            public void run(SourceContext<Integer> sourceContext) throws Exception {
                while (running){
                    sourceContext.collect(random.nextInt(20));
                    TimeUnit.MILLISECONDS.sleep(500L);
                }
            }

            @Override
            public void cancel() {
                running = false;
            }
        }).keyBy(obj->1)
                .process(new IntIncreaseAlert())
                .print();


        env.execute();
    }

    public static class IntIncreaseAlert extends KeyedProcessFunction<Integer,Integer,String> {
        // 上一个数的记录
        private ValueState<Integer> lastInt;
        // 定时器
        private ValueState<Long> timer;

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            lastInt = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("上个数字的记录", Types.INT));
            timer = getRuntimeContext().getState(new ValueStateDescriptor<Long>("定时器", Types.LONG));
        }

        @Override
        public void processElement(Integer value, Context context, Collector<String> out) throws Exception {
            Integer temp = null;
            if (lastInt.value() != null){
                System.out.println("上一个数值："+lastInt.value()+"----本次数值："+value);
                temp = lastInt.value();
            }
            lastInt.update(value);

            Long timeTs = null;
            if (timer.value()  != null){
                timeTs = timer.value();
            }

            if (temp == null || value < temp){
                if (timeTs != null) {
                    context.timerService().deleteProcessingTimeTimer(timeTs);
                    timer.clear();
                }
            }else if (value > temp && timeTs == null){
                long ring = context.timerService().currentProcessingTime() + 1000L;
                context.timerService().registerProcessingTimeTimer(ring);
                timer.update(ring);
            }

        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            super.onTimer(timestamp, ctx, out);
            out.collect("数值上升了！！！");
            timer.clear();
        }
    }

}
